package cn.cas.mango.service;

import cn.cas.mango.Db.ClusterWithScore;
import cn.cas.mango.Db.DbCluster;
import cn.cas.mango.Db.Keywords2Db;
import cn.cas.mango.Db.Keywords2IdDate;
import cn.cas.mango.Db.KeywordsCast;
import cn.cas.mango.Db.KeywordsInsert;
import cn.cas.mango.Db.KeywordsUpdate;
import cn.cas.mango.config.Mapping;
import cn.cas.mango.dbscan.Data;
import cn.cas.mango.dto.param.GlobalParam;
import cn.cas.mango.dto.param.HitNews;
import cn.cas.mango.dto.param.News;
import cn.cas.mango.util.Pair;
import cn.cas.mango.util.TimeUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.client.MongoClients;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;


import java.io.StringReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@Service
@Slf4j
public class PrintJson {
    private final ObjectMapper objectMapper = new ObjectMapper();
    @Autowired
    private GetJson getJson;
    // 最小的keywords大小
    @Value("${mongodb.minimumKeywordsSize}")
    private Integer minimumKeywordsSize;

    @Autowired
    private ElasticsearchClient client;

    @Value("${elasticsearch.newIndex}")
    private String INDEX;
    @Autowired
    private MongoTemplate mongoTemplate;

    private boolean createIndex() {
      boolean exists;
      try {
        exists = client.indices().exists(x -> x.index(INDEX)).value();
        if (exists) {
          client.indices().delete(new DeleteIndexRequest.Builder().index(INDEX).build());
        }
        client.indices().create(x->x.index(INDEX).mappings(m->m.withJson(new StringReader(Mapping.mapping()))));
        return exists;
      } catch (IOException e) {
        throw new RuntimeException(e);
      }

    }

    private File[] getSortedFile(File[] files) {
      List<Pair<String, File>> list = new ArrayList<>();
      for (File file: files) {
        list.add(new Pair<>(file.getName(), file));
      }
      list.sort(Comparator.comparing(Pair::getKey));
      return list.stream().map(Pair::getValue).toArray(File[]::new);
    }

    public void main(GlobalParam globalParam) throws IOException {
        if (createIndex()) {
          log.info("索引存在，已删除后重新建立");
        } else {
          log.info("索引已建立");
        }
      String jsonPath = "json";
      File file1 = new File(jsonPath);
        if (!file1.exists()) {
          file1.mkdir();
          getJson.main(globalParam, jsonPath);
        }

        File dirFile = new File(jsonPath);
        File[] files = getSortedFile(Objects.requireNonNull(dirFile.listFiles()));

        int i = 0;
        TimeUtil timeUtil = new TimeUtil();
        timeUtil.setStart();
        for (File file: files) {
            ObjectMapper mapper = new ObjectMapper();
            List list = mapper.readValue(file, List.class);
            Map<String, News> newsMap = new HashMap<>();
            List<HitNews> hitNewsListBefore = new ArrayList<HitNews>(){{
                for (Object o: list) {
                    News news = JSON.parseObject(JSON.toJSONString(o), News.class);
                    newsMap.put(news.getId(), news);
                    HitNews hitNews = new HitNews(0.0);
                    BeanUtils.copyProperties(news, hitNews);
                    add(hitNews);
                }
            }};
            // 局部去重
          List<HitNews> hitNewsList = removeDuplicates(hitNewsListBefore);
          // 全局去重
          removeDuplicatesInDb(hitNewsList);
          // like
            // cluster_keywords
            // self
          SameNewsService sameNewsService = new SameNewsService(hitNewsList, SortOrder.Desc, "time", 1000);
          Map<String, Map<String, Object>> clusters = sameNewsService
                .getSameNewsList(globalParam.getPartSim());
          Map<String, Map<String, Double>> map2Score = sameNewsService.getMap2Score();
            // 需要更新的id对应簇，key是id，value是属于的簇
            Map<String, String> preparedUpdatedIds = new HashMap<>();
            for (String key: clusters.keySet()) {
              List<String> id2Object = new ArrayList<>();
              id2Object.add(((HitNews)clusters.get(key).get("self")).getId());
              Stream like = ((List) clusters.get(key).get("like")).stream().map(x -> ((HitNews) x).getId());
              id2Object.addAll((List<String>)like.collect(Collectors.toList()));
              for (String id: id2Object) {
                preparedUpdatedIds.put(id, key);
              }
            }
            objectMapper.writeValue(new File("temp.json"),clusters);
            executeMerge(clusters, preparedUpdatedIds, globalParam.getGlobalSim(), file);
            for (String newsId : preparedUpdatedIds.keySet()) {
              String clusterId = preparedUpdatedIds.get(newsId);
              newsMap.get(newsId).setClusterId(clusterId);
            }
            // 写入scoreMap
            write2Db(ClusterWithScore.wrapList(map2Score, newsMap));
            // 写入到es
            List<BulkOperation> operations = newsMap.values().stream().map(news->
                BulkOperation.of(b->b
                    .index(y-> {
                      try {
                        return y
                        .index(INDEX)
                        .id(news.getId())
                        .document(JsonData.fromJson(mapper.writeValueAsString(news)));
                      } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                      }
                    }
                )
            )).collect(Collectors.toList());
            client.bulk(x->x.operations(operations));
            log.info(file.getName() + " finished");
        }
      log.info("total cost " + timeUtil.getCost() + "s");
    }

    private void removeDuplicatesInDb(List<HitNews> hitNewsList) {
      Query query = new Query().addCriteria(Criteria.where("id").in(hitNewsList.stream().map(HitNews::getId).collect(Collectors.toList())));
      HashSet<String> duplicates = mongoTemplate.find(
              query, DbCluster.class, "dbCluster")
          .stream().map(DbCluster::getId).collect(Collectors.toCollection(HashSet::new));
      hitNewsList.removeIf(hitNews -> duplicates.contains(hitNews.getId()));
    }

    private List<HitNews> removeDuplicates(List<HitNews> hitNewsList) {
      Map<String, HitNews> hitNewsMap = new HashMap<String, HitNews>(){{
        for (HitNews hitNews: hitNewsList) {
          put(hitNews.getId(), hitNews);
        }
      }};
      return new ArrayList<>(hitNewsMap.values());
    }

    private void executeMerge(Map<String, Map<String, Object>> clusters,
                              Map<String, String> preparedUpdatedIds, Double globalSim, File file) {
      // 将Cluster包装成DbCluster
      List<DbCluster> dbClusters = wrap(clusters);
      // 需要更新的簇
      List<DbCluster> dbClusters2Update = new ArrayList<>();
      // 需要插入的簇
      List<DbCluster> dbClusters2Insert = new ArrayList<>();
      // 需要更新的关键词
      Map<String, Keywords2Db> keywords2DbMap = new HashMap<>();

      for (DbCluster dbCluster: dbClusters) {
        List<DbCluster> keyWordsCast = findKeyWordsCast(dbCluster, keywords2DbMap);
        DbCluster maxScoreDbCluster = checkNewCluster(keyWordsCast, dbCluster, globalSim);
        if (maxScoreDbCluster != null) {
          // 需要做两方面更新
          // 1. 更新mongodb中该簇的关键词
          Set<String> newKeywords = Data.calInter(dbCluster.getKeywords(), maxScoreDbCluster.getKeywords());
          if (newKeywords.size() < minimumKeywordsSize) {
            for (String keyword: dbCluster.getKeywords()) {
              keywords2DbMap.get(keyword).add(dbCluster.getId(), dbCluster.getDate());
            }
            // 添加该簇到mongodb
            dbClusters2Insert.add(dbCluster);
            continue;
          }
          dbClusters2Update.add(new DbCluster(maxScoreDbCluster.getId(), newKeywords,
              maxScoreDbCluster.getNum() + dbCluster.getNum(), maxScoreDbCluster.getDate()));
          update2Db(dbClusters2Update);
          // 2. 更新preparedUpdatedIds中 该cluster下所有成员归属的簇id
          Set<String> updatingClusterIds = new HashSet<String>(){{
            add(((HitNews) clusters.get(dbCluster.getId()).get("self")).getId());
            addAll(((List<HitNews>)clusters.get(dbCluster.getId()).get("like")).stream().map(News::getId).collect(Collectors.toList()));
          }};

          for (String key: updatingClusterIds) {
            preparedUpdatedIds.put(key, maxScoreDbCluster.getId());
          }
        } else {
          // 更新关键词倒排
          for (String keyword: dbCluster.getKeywords()) {
            keywords2DbMap.get(keyword).add(dbCluster.getId(), dbCluster.getDate());
          }
          // 添加该簇到mongodb
          dbClusters2Insert.add(dbCluster);
        }
      }

      write2Db(dbClusters2Insert);
      // 保存到文件
      File file1 = new File("cache");
      if (!file1.exists()) {
        file1.mkdir();
      }
      try {
        new FileWriter("cache/" + file.getName()).write(dbClusters2Insert.stream().map(DbCluster::toString).collect(Collectors.toList()).toString());
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      // 更新关键词映射
      Collection<Keywords2Db> allKeywords = keywords2DbMap.values();
      List<KeywordsCast> keywordsUpdate = new ArrayList<>();
      List<KeywordsCast> keywordsInsert = new ArrayList<>();
      for (Keywords2Db keywords2Db: allKeywords) {
        if (keywords2Db instanceof KeywordsUpdate) {
          keywordsUpdate.add(keywords2Db.getKeywordsCast());
        } else if (keywords2Db instanceof KeywordsInsert) {
          keywordsInsert.add(keywords2Db.getKeywordsCast());
        }
      }
      update2Db(keywordsUpdate);
      write2Db(keywordsInsert);
    }
    // 找到关键词对应的所有簇
    private List<DbCluster> findKeyWordsCast(DbCluster dbCluster, Map<String, Keywords2Db> keywords2DbMap) {
      Set<String> queryKeywords = dbCluster.getKeywords();
      if (queryKeywords.size() == 0) {
        return new ArrayList<>();
      }
      Query query = new Query().addCriteria(Criteria.where("id").in(queryKeywords));
      List<KeywordsCast> keywordsCasts = mongoTemplate.find(query, KeywordsCast.class, "keywordsCast");
      Map<String, KeywordsCast> keywordsCastMap = new HashMap<>();
      for (KeywordsCast keywordsCast: keywordsCasts) {
        keywordsCastMap.put(keywordsCast.getId(), keywordsCast);
      }
      Set<String> updateKeywords = new HashSet<>(keywordsCastMap.keySet());
      for (String queryKeyword: queryKeywords) {
        if (!keywords2DbMap.containsKey(queryKeyword)) {
          if (updateKeywords.contains(queryKeyword)) {
            keywords2DbMap.put(queryKeyword, new KeywordsUpdate(queryKeyword, dbCluster.getId(), dbCluster.getDate()));
          } else {
            keywords2DbMap.put(queryKeyword, new KeywordsInsert(queryKeyword, dbCluster.getId(), dbCluster.getDate()));
          }
        }
      }
      List<String> allDocumentsIds = new ArrayList<>();
      for (KeywordsCast keywordsCast: keywordsCasts) {
        allDocumentsIds.addAll(keywordsCast.getDocument().stream().map(Keywords2IdDate::getId).collect(Collectors.toList()));
      }
      return mongoTemplate.find(new Query().addCriteria(Criteria.where("id").in(allDocumentsIds)), DbCluster.class, "dbCluster");
    }

    // 检查现有库，如果可以合并，那么合并
    private DbCluster checkNewCluster(List<DbCluster> dbClusters, DbCluster comparedDbCluster, double limit) {
      List<Pair<String, Double>> dbList = new ArrayList<>();
      Map<String, DbCluster> dbClusterMap = new HashMap<String, DbCluster>(){{
        for(DbCluster dbCluster: dbClusters) {
          put(dbCluster.getId(), dbCluster);
        }
      }};
      for (DbCluster dbCluster: dbClusters) {
        dbList.add(new Pair<>(dbCluster.getId(), new Data(dbCluster.getId(), dbCluster.getKeywords())
            .calSim(new Data(comparedDbCluster.getId(), comparedDbCluster.getKeywords()))));
      }
      dbList.sort((o1, o2)-> {
        if (o1 == null || o2 == null) {
          if (o1 == o2) {
            return 0;
          } else if (o1 == null) {
            return 1;
          } else {
            return -1;
          }
        }
        return -Double.compare(o1.getValue(), o2.getValue());
      });
      if (dbList.size() == 0 || dbList.get(0).getValue() < limit) {
        return null;
      }
      return dbClusterMap.get(dbList.get(0).getKey());
    }

    private Date transdate(String date) {
      SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
      try {
        return sdf.parse(date);
      } catch (ParseException e) {
        throw new RuntimeException(e);
      }
    }
    List<DbCluster> wrap(Map<String, Map<String, Object>> clusters) {
        List<DbCluster> dbClusters = new ArrayList<>();
        for (String id: clusters.keySet()) {
            Map<String, Object> info = clusters.get(id);
            Set<String> keywords = (Set<String> ) info.get("cluster_keywords");
            int num = ((List) info.get("like")).size() + 1;
            DbCluster dbCluster = new DbCluster(id, keywords, num, transdate(((HitNews) info.get("self")).getNewsPublicDate()));
            dbClusters.add(dbCluster);
        }
        return dbClusters;
    }

    void update2Db(List<?> list) {
      if (list.size() == 0) return;
      Object o = list.get(0);
      if (o instanceof DbCluster) {
        BulkOperations bulkOperations = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, DbCluster.class);
        for (Object o1: list) {
          DbCluster db = (DbCluster) o1;
          Update update = new Update()
              .set("keywords", db.getKeywords())
              .set("num", db.getNum())
              .set("date", db.getDate());
          bulkOperations.updateOne(Query.query(Criteria.where("id").is(db.getId())), update);
        }
        bulkOperations.execute();
      } else if (o instanceof KeywordsCast) {
        BulkOperations bulkOperations = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, KeywordsCast.class);
        for (Object o1: list) {
          KeywordsCast keywordsCast = (KeywordsCast) o1;
          Update update = new Update().push("document").each(keywordsCast.getDocument());
          bulkOperations.updateOne(Query.query(Criteria.where("id").is(keywordsCast.getId())), update);
        }
        bulkOperations.execute();
      }
      list.clear();
    }
    void write2Db(List<?> list) {
      if (list.size() == 0) return;
      Object o = list.get(0);
      if (o instanceof DbCluster) {
        mongoTemplate.insert(list, DbCluster.class);
      } else if (o instanceof KeywordsCast){
        mongoTemplate.insert(list, KeywordsCast.class);
      } else if (o instanceof ClusterWithScore) {
        mongoTemplate.insert(list, ClusterWithScore.class);
      }
    }


}
